package com.qidian.common.beanCache;

import java.util.Collection;
import java.util.HashMap;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.atomic.AtomicInteger;
//import java.util.concurrent.Co

import com.qidian.client.TransConstants;
import com.qidian.common.Constants;
import com.qidian.common.MessageStatus;
import com.qidian.common.TransType;
import com.qidian.common.beans.GameTypeBean;
import com.qidian.common.beans.MessageBean;



/**
 *  用于缓存消息。
 *  
 *  当插件发送消息过来时，将消息插入缓存；
 *  当答题客户端提交答案时，更新缓存中的消息；
 *  当插件提取答案后，更新消息；
 *  清理工线程，将缓存中过期的、已经应答过的消息保存在数据库中，并将它们从缓存中删除；
 * 
 * */
public final class MessageBeanCache
{
	private static MessageBeanCache messageBeanCache = new MessageBeanCache() ;
	
    /*  缓冲的最多消息数量*/
    public final static int max_buf_size = 10000 ;
    
    /* 初始的缓冲大小  */
    public static int init_buf_size = 8192 ;    
    
    /*  取题次数	*/
    public static int fetchCount = 0 ;
    public static int second_rate = 3 ;
    public static int third_rate = 5 ;
    public AtomicInteger waitMsgSize = new AtomicInteger(0) ;
    
    /*	<消息ID，消息>	*/
    public ConcurrentHashMap<Long,MessageBean> buffer = null ;
	
    //	代分配的消息队列，三个优先级
    // 10-30,紧急
	private ConcurrentLinkedQueue<MessageBean> waitMsgQueue_first = null ;
	// 30-70,中等
	private ConcurrentLinkedQueue<MessageBean> waitMsgQueue_second = null ;
	// 70+,普通
	private ConcurrentLinkedQueue<MessageBean> waitMsgQueue_third = null ;

	// 转发队列长度
    public AtomicInteger transMsgSize_microRui = new AtomicInteger(0) ;
    public AtomicInteger transMsgSize_uu_lz = new AtomicInteger(0) ; 
	// 分流转发队列 微锐
	private ConcurrentLinkedQueue<MessageBean> transMsgQueue_microRui = null ;
	// 分流转发队列 UU 联众
	private ConcurrentLinkedQueue<MessageBean> transMsgQueue_uu_lz = null ;
	// 转发队列<转发平台编号,转发队列>
	private HashMap<TransType,ConcurrentLinkedQueue<MessageBean>> transMsgQueues = null ;
	// 转发队列<转发平台编号,转发队列长度>	
	private HashMap<TransType,AtomicInteger> transMsgSizes = null ;
	
	/*	缓冲的消息状态计数	*/
	public MessageStatusCounter mscounter = new MessageStatusCounter() ;
	
    private MessageBeanCache()
    {
    	buffer = new ConcurrentHashMap<Long,MessageBean>(init_buf_size) ;
    	waitMsgQueue_first = new ConcurrentLinkedQueue<MessageBean>() ;
    	waitMsgQueue_second = new ConcurrentLinkedQueue<MessageBean>() ;
    	waitMsgQueue_third = new ConcurrentLinkedQueue<MessageBean>() ;
    	
    	// 转发队列初始化
    	transMsgQueue_microRui = new ConcurrentLinkedQueue<MessageBean>() ;
    	transMsgQueue_uu_lz = new ConcurrentLinkedQueue<MessageBean>() ;
    	transMsgQueues = new HashMap<TransType,ConcurrentLinkedQueue<MessageBean>>() ;
    	transMsgQueues.put(TransType.TRANS_MICRORUI, transMsgQueue_microRui) ;
    	transMsgQueues.put(TransType.TRANS_LZ, transMsgQueue_uu_lz) ;
    	transMsgSizes = new HashMap<TransType,AtomicInteger>() ;
    	transMsgSizes.put(TransType.TRANS_MICRORUI, transMsgSize_microRui) ;
    	transMsgSizes.put(TransType.TRANS_LZ, transMsgSize_uu_lz) ;
    }
    
    public static MessageBeanCache getInstance()
    {
    	return messageBeanCache ;
    }
    /**
     * 根据ID返回消息
     * */
    public MessageBean get(Long id)
    {
        return buffer.get(id) ;
    }
    
    /**
     * 
     * 从一个转发队列中获取一个消息
     * 
     * 
     * */
    private MessageBean poll(ConcurrentLinkedQueue<MessageBean> queue, AtomicInteger waitMsgSize)
    {
    	MessageBean msg = queue.poll() ;
    	while(msg != null )
    	{
    		waitMsgSize.addAndGet(-1) ;
    		if( msg.isTimeOut())
    		{
    			msg.setStatus(MessageStatus.TIME_OUT.status) ;
    			msg = queue.poll() ;
    		}
    		else
    			return msg ;
    	}
    	return null ;
    }
    /**
     * 返回下一个代处理的消息,
     * 过滤超时的消息；
     * 
     * */
    public MessageBean poll()
    {

    	MessageBean msg = null ;
    
    	// 防止饿死,每5次从第二队列取一次
    	fetchCount ++ ;
    	if(fetchCount%second_rate == 0)
    	{
    		msg = poll(waitMsgQueue_second,waitMsgSize) ;
        	if (msg != null)
        		return msg ;
    	} 
    	// 防止饿死,每7次从第三队列取一次
    	if(fetchCount%third_rate == 0)
    	{
    		msg = poll(waitMsgQueue_third,waitMsgSize) ;
        	if (msg != null)
        		return msg ;
    	}
    	
    	// 首先从第一队列获取
    	msg = poll(waitMsgQueue_first,waitMsgSize) ;
    	if (msg != null)
    		return msg ;
    	
    	// 从第二队列获取
    	msg = poll(waitMsgQueue_second,waitMsgSize) ;
    	if (msg != null)
    		return msg ;
    	
    	// 从第三队列获取
    	msg = poll(waitMsgQueue_third,waitMsgSize) ;
    	if (msg != null)
    		return msg ;
    	
    	// 从转发队列获取
    	return pollTrans() ;
    }
    
    /**
     * 从转发队列获取下一个待处理的消息
     * 
     * */
    public MessageBean pollTrans()
    {
    	MessageBean msg =  null;
    	
    	// 首先从微锐获取
    	msg = poll(transMsgQueue_microRui,transMsgSize_microRui) ;
    	if (msg != null)
    		return msg ;
    	
    	// 再从UU获取
    	msg = poll(transMsgQueue_uu_lz,transMsgSize_uu_lz) ;
    	
    	return msg ;
    }
    
    /**
     * 从制定转发队列获取下一个待处理的消息
     * 
     * */
    public MessageBean pollTrans(int transNo)
    {
    	
    	if (transNo <= 0)
    		return pollTrans() ;
    	
    	ConcurrentLinkedQueue<MessageBean> transMsgQueue = transMsgQueues.get(TransType.getTransType(transNo)) ;
    	if (transMsgQueue == null)
    	{
    		System.out.println("无效的转发ID");
    		return null ;
    	}
    	AtomicInteger transMsgSize = transMsgSizes.get(TransType.getTransType(transNo)) ;
    	
    	return poll(transMsgQueue,transMsgSize) ;
    	
    }
    
    /**
     * 添加消息到制定队列
     * 
     * 
     * */
    public boolean add(MessageBean message,GameTypeBean gameType)
    {
        /*  如果缓冲太多消息，丢弃之    */
        if (buffer.size() > max_buf_size)
            return false;
        
        buffer.put(message.getId(), message) ;
        
        // 如果当前服务器队列长度超过阈值，并且题目不参与转发
        if (waitMsgSize.get() < (int)(gameType.getDifficulty()*Constants.trans_threshold) 
        		|| !gameType.isTransOn())
        {
        	waitMsgSize.incrementAndGet() ;
        	if (gameType.getMinInterval() <= Constants.interval_split_1)
        		waitMsgQueue_first.add(message) ;
        	else if (gameType.getMinInterval() <= Constants.interval_split_2)
        		waitMsgQueue_second.add(message) ;
        	else
        		waitMsgQueue_third.add(message) ;
        }
        else
        {
        	transMsgQueues.get(gameType.getTransOn()).add(message) ;
        	transMsgSizes.get(gameType.getTransOn()).incrementAndGet() ;
        }
        
        return true ;
    }
    
    public void remove(Long id)
    {
        buffer.remove(id) ;
    }
    
    /**
     * 缓存消息数量
     * 
     * */
    public int size()
    {
        return buffer.size() ;
    }
    
    public void print()
    {
    	System.out.println("Cached message count:"+buffer.size()) ;
    	Collection<MessageBean> values = buffer.values() ;
    	for(MessageBean bean: values)
    	{
    		System.out.println(bean) ;
    	}
    }
	
    /**
     * 服务器队列长度：包括转发队列
     * 
     * */
	public int getWaitSize() 
	{
		return waitMsgSize.get() + transMsgSize_microRui.get();
	}
	/**
	 * 获取第一队列长度
	 * 
	 * */
	public int getFirstWaitSize()
	{
		return waitMsgQueue_first.size() ;
	}
	/**
	 * 
	 * 获取第二队列长度
	 * 
	 * 
	 * */
	public int getSecondWaitSize()
	{
		return waitMsgQueue_second.size() ;
	}
	/**
	 * 获取第三队列长度
	 * 
	 * */
	public int getThirdWaitSize()
	{
		return waitMsgQueue_third.size() ;
	}
	/**
	 * 待转发队列总长度
	 * 
	 * */
	public int getTransSize()
	{
		return transMsgSize_microRui.get() + transMsgSize_uu_lz.get() ;
	}
	/**
	 * 特定待转发队列长度
	 * 
	 * */
	public int getTransSize(int transNo)
	{
		return transMsgSizes.get(TransType.getTransType(transNo)).get() ;
	}
	public String toString()
	{
		StringBuffer string = new StringBuffer(256) ;
		string.append("Buf size:"+buffer.size()+",wait queue size:"+waitMsgSize.get()
				+", trans queue size:" + transMsgSize_microRui.get());
		return string.toString() ;
	}
}

